package com.jch.dm5_producer_consumer.reusable_code.work_stealing;

import com.jch.dm3_two_phase_termination.reusable_code.AbstractTerminatableThread;
import com.jch.dm3_two_phase_termination.reusable_code.TerminationToken;

import java.util.Random;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;

public class WorkStealingExample {

    private final WorkStealingEnabledChannel<String> channel;
    private final TerminationToken token = new TerminationToken();

    public WorkStealingExample() {
        int nCPU = Runtime.getRuntime().availableProcessors();
        int consumerCount = nCPU / 2 + 1;

        @SuppressWarnings("unchecked")
        BlockingDeque<String>[] managedQueues = new LinkedBlockingDeque[consumerCount];

        // 该通道实例对应了多个队列实例managedQueues
        channel = new WorkStealingChannel<String>(managedQueues);

        Consumer[] consumers = new Consumer[consumerCount];
        for (int i = 0; i < consumerCount; i++) {
            managedQueues[i] = new LinkedBlockingDeque<String>();
            consumers[i] = new Consumer(token, managedQueues[i]);
        }
        for (int i = 0; i < nCPU; i++) {
            new Producer().start();
        }
        for (int i = 0; i < consumerCount; i++) {
            consumers[i].start();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        WorkStealingExample example = new WorkStealingExample();
        Thread.sleep(3000);
    }

    private class Producer extends AbstractTerminatableThread {

        private int i = 0;

        @Override
        protected void doRun() throws Exception {
            channel.put(String.valueOf(i++));
            token.reservations.incrementAndGet();
        }
    }

    private class Consumer extends AbstractTerminatableThread {

        private final BlockingDeque<String> workQueue;

        public Consumer(TerminationToken terminationToken, BlockingDeque<String> workQueue) {
            super(terminationToken);
            this.workQueue = workQueue;
        }

        @Override
        protected void doRun() throws Exception {
            String product = channel.take(workQueue);
            System.out.println("Processing product:" + product);
            try {
                // 模拟操作的时间消耗
                Thread.sleep(new Random().nextInt(100));
            } catch (InterruptedException e) {
            } finally {
                token.reservations.decrementAndGet();
            }
        }
    }
}
